package MREngine;

import x10.util.*;

public class MapReduceEngineSeq[K, V] {
	
	/**
	 * This list is a two levle list. 
	 * The first level is a list of ShuffledListEntry, indexed by K.
	 * The second level is a list of V.
	 */
	// private static class ShuffledList[K, V]
	// {
	// 	private static class ShuffledListEntry[K, V]
	// 	{
	// 		private val key: K;
	// 		private var val_list: ArrayList[MREEntry[K, V]];
	// 		
	// 		public def this(_key:K)
	// 		{
	// 			key = _key;
	// 			val_list = new ArrayList[MREEntry[K, V]]();
	// 		}
	// 		
	// 		public def this(_key:K, _value:V)
	// 		{
	// 			key = _key;
	// 			val new_entry = new MREEntry[K, V](_key, _value);
	// 			val_list = new ArrayList[MREEntry[K, V]]();
	// 			//val_list.add(new_entry);
	// 		}
	// 		
	// 		public def this(_entry: MREEntry[K, V])
	// 		{
	// 			key = _entry.getKey();
	// 			val_list = new ArrayList[MREEntry[K, V]]();
	// 			//val_list.add(_entry);
	// 		}
	// 		
	// 		public def getKey() = key;
	// 		public def getValList() = val_list;
	// 		
	// 		/**
	// 		 * add V to the val_list
	// 		 */
	// 		public def add (_entry: MREEntry[K, V])
	// 		{
	// 			return val_list.add(_entry);
	// 		}
	// 	}
	// 	
	// 	private var key_list: ArrayList[ShuffledListEntry[K, V]];
	// 	private var count_all:Int;
	// 	
	// 	
	// 	public def this()
	// 	{
	// 		var db:MREData = new MREData();
	// 		db.getLabel();
	// 		count_all = 0;
	// 		key_list = new ArrayList[ShuffledListEntry[K, V]](); 
	// 	}
	// 	
	// 	/**
	// 	 * To find the shuffleListEntry with the K in the shuffledList
	// 	 */
	// 	private def find_by_key (_key:K) : ShuffledListEntry[K, V]
	// 	{
	// 		if (_key == null)
	// 			return null;
	// 		var sle: ShuffledListEntry[K, V];
	// 		for (var i: Int = 0; i < key_list.size(); i++) 
	// 		{
	// 			sle = key_list.get(i);
	// 			if (_key.equals(sle.getKey()))
	// 				return sle;
	// 		}
	// 		
	// 		return null;
	// 	}
	// 	
	// 	/**
	// 	 * To add a (K, V) pair to the shuffled list
	// 	 */
	// 	public def add (entry:MREEntry[K, V]): Boolean
	// 	{
	// 		var sle:ShuffledListEntry[K, V] = find_by_key (entry.getKey());
	// 		if (sle == null)
	// 		{
	// 			sle = new ShuffledListEntry[K, V](entry);
	// 			key_list.add(sle);
	// 		}
	// 		
	// 		count_all++;
	// 		return sle.add(entry);
	// 	}
	// 	
	// 	public def size1 () = key_list.size();
	// 	public def size_all () = count_all; 
	// 	public def getValList(i:Int) = key_list.get(i).getValList(); 
	// 	public def getKey(i:Int) = key_list.get(i).getKey();
	// }
	
	/** the raw data to be processed */
	val data:MREData;
	
	/** function pointers of split, map, reduce, and print functions */
	private var split_func: (data:MREData) => void;
	private var map_func: (data:MREData) => void;
	private var reduce_func: (key:K, value:ArrayList[MREEntry[K, V]]) => void;
	//private var print_func: (key:K, value:V) => void;
	
	/** the list of splitted datablocks */
	private var split_list: ArrayList[MREData];
	
	/** the list of mapped (K V) pairs */
	private var intermediate_list: ArrayList[MREEntry[K,V]];
	
	/** the hashmap of shuffled (K V) pair indexed by K */
	private var shuffled_hashmap: HashMap[K, ArrayList[MREEntry[K, V]]];
	
	/** The array to hold the list of each K for reducing.
	 * Array Size: number of K.
	 * init before use.
	 */
	private var shuffled_array: Array[ArrayList[MREEntry[K,V]]];
	
	/** The hash map to save the final result. 
	 * HashMap Size: number of K.
	 */
	private var final_hashmap:HashMap[K, MREEntry[K, V]];
	
	/** the list of final results */
	private var final_list: ArrayList[MREEntry[K, V]];
	
	private var ready:Boolean;
	
	/**
	 * The constructor function
	 * init_data: the initial data block to do the map-reduce
	 */
	public def this (_init_data:MREData)
	{
		data = _init_data;
		ready = false;
	}
	
	private def initialize(): void
	{
		// initialize the split_list
		split_list = new ArrayList[MREData]();
		
		// initialize the map_list
		intermediate_list = new ArrayList[MREEntry[K,V]]();
		
		// initialize the shuffled_hashmap
		shuffled_hashmap = new HashMap[K, ArrayList[MREEntry[K, V]]]();
		
		// init the final_hashmap
		final_hashmap = new HashMap[K, MREEntry[K, V]]();
		
		// initialize the final_list
		final_list = new ArrayList[MREEntry[K,V]]();
		
		ready = true;
	}
	
	private def clear(): void
	{
		split_list = null;
		intermediate_list = null;
		shuffled_hashmap = null;
		//final_list = null;
		ready = false;
	}
	
	private def shuffle (_entry:MREEntry[K, V])
	{
		var tmpBox:Box[ArrayList[MREEntry[K, V]]] = shuffled_hashmap.get(_entry.getKey());
		var tmpArrList:ArrayList[MREEntry[K, V]];
		//Console.OUT.println("DEBUG: tmpArrList == null: " + tmpBox == null);
		if (tmpBox == null)		// if not found in hash map, create a new ArrayList
		{
			tmpArrList = new ArrayList[MREEntry[K, V]]();
			shuffled_hashmap.put(_entry.getKey(), tmpArrList);
		}
		else
			tmpArrList = tmpBox.value;
		
		val ret = tmpArrList.add(_entry);
		//Console.OUT.println("DEBUG shuffle return: " + ret);
	}
	
	/**
	 * Call by the user's split function.
	 * Add a split of data to the data split list.
	 */
	public def EmitSplit (_data: MREData): void
	{
		split_list.add (_data);
	}
	
	/**
	 * Call by the user's map function.
	 * Add a mapped (K V) pair to the intermediate list.
	 */
	public def EmitIntermediate (_key:K, _value:V): void
	{
		val new_intermediate = new MREEntry[K,V](_key, _value);
		//Console.OUT.println("DEBUG _key= "+_key +"; value= " + _value + "; add = " +intermediate_list.add(new_intermediate));
		intermediate_list.add(new_intermediate);
	}
	
	/**
	 * Call by the user's reduce function.
	 * Add a final (K V) pair to the final list.
	 */
	public def EmitFinal (_key:K, _value:V): void
	{
		val new_final = new MREEntry[K,V](_key, _value);
		final_list.add(new_final);
		final_hashmap.put(_key, new_final);
	}
	
	/**
	 * Find _key in the final_hashmap.
	 * Retrun the (K, V) entry if found.
	 * Otherwise, return null.
	 */
	public def FindByKey (_key: K): MREEntry[K, V]
	{
		return final_hashmap.get(_key) != null ?
				final_hashmap.get(_key).value
				: null;
	}
	
	public def GetResults (): ArrayList[MREEntry[K,V]] = final_list;
	
	/**
	 * The set functions for three user-defined functions, 
	 * split, map, and reduce.
	 */
	public def SetSplitFunc (_func: (data:MREData) => void): void
	{
		split_func = _func;
	}
	
	public def SetMapFunc (_func: (data:MREData) => void): void
	{
		map_func = _func;
	}
	
	public def SetReduceFunc (_func: (key:K, value:ArrayList[MREEntry[K, V]]) => void): void
	{
		reduce_func = _func;
	}
	
	/** 
	 * the sequential version of the actual scheduler. 
	 */
	public def MapReduceSchedulerSerial(): void
	{
		if (!ready)
			initialize();
		
		if (data == null)
			return;
		
		// split the raw data
		split_func (data);
		
		//Console.OUT.println("DEBUG split_list size: " + split_list.size());
		
		// map the data to (K V) pairs
		for (i in (0..(split_list.size()-1)))
		{
			//Console.OUT.println("DEBUG: i = ["+ i +"]");
			map_func (split_list.get(i));
		}
		//Console.OUT.println("DEBUG intermediate_list size: " + intermediate_list.size());
		
		
		// shuffle the (K V) pairs
		for (i in (0..(intermediate_list.size()-1)))
		{
			shuffle (intermediate_list.get(i));
		}
		//Console.OUT.println("DEBUG shuffled_hashmap size: " + shuffled_hashmap.size());
		
		// reduce the (K V) pairs to final result
		var set:Set[Map.Entry[K,ArrayList[MREEntry[K,V]]]] = shuffled_hashmap.entries();
		i:Iterator[Map.Entry[K,ArrayList[MREEntry[K,V]]]] = set.iterator();
		shuffled_array= new Array[ArrayList[MREEntry[K,V]]](set.size());
		
		var count:Int = 0;
		while(i.hasNext())
		{
			var value:ArrayList[MREEntry[K, V]]= i.next().getValue();
			shuffled_array(count) = value;
			count++;
		}
		
		for (j in (0..(shuffled_array.size-1)))
		{
			val key:K = shuffled_array(j).get(0).getKey();
			//Console.OUT.println("DEBUG iteration: " + j);
			//Console.OUT.println("DEBUG shuffle vallist size: " + shuffled_array(j).size());
			reduce_func (key, shuffled_array(j));
		}
		
		clear();
		
		//Console.OUT.println("DEBUG final_list size: " + final_list.size());
	}

}
